This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

Moving back to draft mode. Thanks for taking a look.

This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

End-to-end test: mysql_cdc enterprise connector

Validated the Pipeline CRD against MySQL with binlog CDC streaming into Redpanda. Used the merged-with-main branch tip (340a0eb7) which includes the auto-license-propagation fix.

Stack

• Cluster: k3d v5.8.3 with rancher/k3s:v1.32.13-k3s1
• Operator: built from this branch as localhost/redpanda-operator:dev, --enable-connect=true, enterprise.licenseSecretRef pointing at a CONNECT-enabled enterprise license
• Redpanda: v26.1.1 cluster (single broker, unauthenticated)
• MySQL: 8.0 with --log-bin --binlog-format=ROW --binlog-row-image=FULL --gtid-mode=ON --enforce-gtid-consistency=ON
• Pipeline CR: mysql_cdc input → redpanda output (using clusterRef)

Result: ✅ Pass

• Pipeline phase reached Running (Ready=True, ConfigValid=True, ClusterRefResolved=True)
• All 5 snapshot rows landed on topic mysql.shop.orders (offsets 0–4)
• 2 rows inserted live after pipeline start propagated through binlog and arrived at offsets 5–6
• Auto-rendered mysql-cdc-orders-license Secret was owned by the Pipeline CR (cleaned up on kubectl delete pipeline)
• REDPANDA_LICENSE env var sourced via secretKeyRef to the auto-rendered Secret — no manual license wiring needed

{
  "topic": "mysql.shop.orders",
  "key": "orders",
  "value": "{\"cdc_received_at\":\"...\",\"created_at\":\"...\",\"customer\":\"alice\",\"id\":1,\"product\":\"wrench\",\"qty\":3}",
  "offset": 0
}

Reproduction steps

1. Prerequisites

• nix develop shell (provides k3d, kubectl, helm)
• Docker / Rancher Desktop running
• A Redpanda enterprise license file that includes the CONNECT product (free trials don't qualify — both the operator's gate and the Connect runtime's gate require it)

2. Build the operator image

BUILD_GOOS=linux BUILD_GOARCH=arm64 task build:operator-image \
  PLATFORMS=linux/arm64 \
  CLI_ARGS=--load

Produces localhost/redpanda-operator:dev.

3. Spin up a k3d cluster + cert-manager + operator

k3d cluster create pipeline-e2e \
  --image rancher/k3s:v1.32.13-k3s1 --no-lb \
  --k3s-arg '--disable=traefik@server:0' \
  --k3s-arg '--disable=metrics-server@server:0'

k3d image import localhost/redpanda-operator:dev --cluster pipeline-e2e

helm repo add jetstack https://charts.jetstack.io --force-update
helm install cert-manager jetstack/cert-manager \
  -n cert-manager --create-namespace \
  --version v1.17.2 --set crds.enabled=true --wait

kubectl create namespace redpanda
kubectl create namespace mysql

kubectl -n redpanda create secret generic redpanda-license \
  --from-file=license=/path/to/redpanda.license

helm install redpanda-operator ./operator/chart \
  -n redpanda \
  --set image.repository=localhost/redpanda-operator \
  --set image.tag=dev \
  --set image.pullPolicy=IfNotPresent \
  --set rbac.createAdditionalControllerCRs=false \
  --set connectController.enabled=true \
  --set enterprise.licenseSecretRef.name=redpanda-license \
  --set enterprise.licenseSecretRef.key=license \
  --set crds.enabled=true \
  --set crds.experimental=true \
  --wait

4. Deploy a Redpanda cluster

# redpanda.yaml
apiVersion: cluster.redpanda.com/v1alpha2
kind: Redpanda
metadata:
  name: redpanda
  namespace: redpanda
spec:
  chartRef: {}
  clusterSpec:
    statefulset:
      replicas: 1
      sideCars:
        controllers:
          enabled: true
    resources:
      cpu: { cores: 1 }
      memory: { container: { max: 2Gi } }
    config:
      cluster:
        auto_create_topics_enabled: true
    console: { enabled: false }
    tls: { enabled: false }
    listeners:
      kafka:          { tls: { enabled: false } }
      admin:          { tls: { enabled: false } }
      schemaRegistry: { tls: { enabled: false } }
      http:           { tls: { enabled: false } }

kubectl apply -f redpanda.yaml

5. Deploy MySQL with binlog enabled and seed data

# mysql.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: mysql-init
  namespace: mysql
data:
  init.sql: |
    CREATE DATABASE IF NOT EXISTS shop;
    USE shop;
    CREATE TABLE IF NOT EXISTS orders (
      id INT AUTO_INCREMENT PRIMARY KEY,
      customer VARCHAR(64) NOT NULL,
      product  VARCHAR(64) NOT NULL,
      qty      INT          NOT NULL,
      created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
    INSERT INTO orders (customer, product, qty) VALUES
      ('alice','wrench',3),('bob','hammer',1),('carol','sprocket',7),
      ('dave','wrench',2),('eve','cog',5);
    CREATE USER IF NOT EXISTS 'cdc'@'%'
      IDENTIFIED WITH mysql_native_password BY 'cdcpass';
    GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT, RELOAD, LOCK TABLES,
          SHOW VIEW, PROCESS ON *.* TO 'cdc'@'%';
    FLUSH PRIVILEGES;
---
apiVersion: v1
kind: Service
metadata: { name: mysql, namespace: mysql }
spec:
  selector: { app: mysql }
  ports: [{ port: 3306, targetPort: 3306 }]
---
apiVersion: apps/v1
kind: StatefulSet
metadata: { name: mysql, namespace: mysql }
spec:
  serviceName: mysql
  replicas: 1
  selector: { matchLabels: { app: mysql } }
  template:
    metadata: { labels: { app: mysql } }
    spec:
      containers:
        - name: mysql
          image: mysql:8.0
          env:
            - { name: MYSQL_ROOT_PASSWORD, value: rootpass }
            - { name: MYSQL_DATABASE,      value: shop }
          args:
            - --server-id=1
            - --log-bin=/var/lib/mysql/mysql-bin
            - --binlog-format=ROW
            - --binlog-row-image=FULL
            - --gtid-mode=ON
            - --enforce-gtid-consistency=ON
            - --default-authentication-plugin=mysql_native_password
          ports: [{ name: mysql, containerPort: 3306 }]
          volumeMounts:
            - { name: init, mountPath: /docker-entrypoint-initdb.d }
            - { name: data, mountPath: /var/lib/mysql }
          readinessProbe:
            tcpSocket: { port: 3306 }
            initialDelaySeconds: 15
            periodSeconds: 5
      volumes:
        - { name: init, configMap: { name: mysql-init } }
        - { name: data, emptyDir: {} }

kubectl apply -f mysql.yaml

6. Apply the Pipeline CR

The user-managed Secret holds only the DSN — the operator auto-mounts the license:

# pipeline.yaml
apiVersion: v1
kind: Secret
metadata: { name: mysql-cdc-secret, namespace: redpanda }
stringData:
  MYSQL_DSN: "cdc:cdcpass@tcp(mysql.mysql.svc.cluster.local:3306)/shop"
---
apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata: { name: mysql-cdc-orders, namespace: redpanda }
spec:
  cluster:
    clusterRef: { name: redpanda }
  secretRef:
    - name: mysql-cdc-secret
  configYaml: |
    input:
      mysql_cdc:
        dsn: "${MYSQL_DSN}"
        stream_snapshot: true
        snapshot_max_batch_size: 100
        tables:
          - orders
        flavor: mysql
        checkpoint_cache: mysql_cdc_orders_checkpoint
        checkpoint_key: "mysql_cdc_orders_checkpoint"
        checkpoint_limit: 1024
    pipeline:
      processors:
        - mapping: |
            root = this
            root.cdc_received_at = now()
    output:
      redpanda:
        seed_brokers: [ "${RPK_BROKERS}" ]
        topic: "mysql.shop.orders"
        key: '${! @table }'
    cache_resources:
      - label: mysql_cdc_orders_checkpoint
        memory: { default_ttl: 1h }
  replicas: 1

kubectl apply -f pipeline.yaml

Notes on the mysql_cdc config:

• tables takes unqualified names (the DSN's /shop already scopes the database). shop.orders is rejected as invalid table name.
• checkpoint_cache is required — needs to reference one of the cache_resources labels.
• RPK_BROKERS is injected by the operator from the clusterRef resolution.

7. Verify

kubectl -n redpanda get pipeline mysql-cdc-orders
# NAME               READY   PHASE     REPLICAS   AVAILABLE   AGE
# mysql-cdc-orders   True    Running   1          1           ...

kubectl -n redpanda exec redpanda-0 -c redpanda -- \
  rpk topic consume mysql.shop.orders \
  --brokers redpanda-0.redpanda.redpanda.svc.cluster.local:9093 \
  -n 5 --offset start

# Insert live rows and watch them stream:
kubectl -n mysql exec mysql-0 -- mysql -uroot -prootpass -e \
  "USE shop; INSERT INTO orders (customer,product,qty) VALUES ('frank','gear',11);"

kubectl -n redpanda exec redpanda-0 -c redpanda -- \
  rpk topic consume mysql.shop.orders \
  --brokers redpanda-0.redpanda.redpanda.svc.cluster.local:9093 \
  -n 1 --offset 5

8. Cleanup

k3d cluster delete pipeline-e2e

Going to mark ready for review given was able to run an end to end with a few connectors successfully

This PR is stale because it has been open 5 days with no activity. Remove stale label or comment or this will be closed in 5 days.

End-to-end test: EKS + RDS, new spec shape, Redpanda Connect v4.92.0

Re-ran the mysql_cdc-into-Redpanda scenario from earlier, this time against real AWS infra and using the v2 Pipeline spec from e7a2da66 (cluster.clusterRef + userRef + valueSources) instead of the old secretRef[]/env[] shape.

Stack

Result: ✅ Pass

• Pipeline Phase=Running, Ready=True, ConfigValid=True, ClusterRef=True (ClusterRefResolved), UserRef=True (UserResolved with mechanism SCRAM-SHA-512).
• All 5 snapshot rows reached mysql.shop.orders (offsets 0–4).
• One live-INSERT row reached the topic at offset 10 (offsets 5–9 are a second snapshot from a pipeline restart during the operator config dance; that's a separate gotcha, see "Notes" below).

offset=0   customer=alice    product=wrench    qty=3   id=1   snapshot
offset=1   customer=carol    product=sprocket  qty=7   id=3   snapshot
offset=2   customer=dave     product=wrench    qty=2   id=4   snapshot
offset=3   customer=bob      product=hammer    qty=1   id=2   snapshot
offset=4   customer=eve      product=cog       qty=5   id=5   snapshot
offset=5–9                                                    re-snapshot after pipeline pod restart
offset=10  customer=frank    product=gear      qty=11  id=6   live binlog replay  ← validates CDC

What the new spec actually buys you

The CR for the pipeline has no inline brokers / TLS / SASL credentials — the operator generates the top-level redpanda block from cluster.clusterRef + userRef:

Pipeline CR (excerpt):

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata: { name: mysql-cdc-orders, namespace: redpanda }
spec:
  # No `image:` here — the chart-level default `connectController.image`
  # set during `helm install` (step 5 below) pins this Pipeline (and every
  # other Pipeline in the cluster) to redpandadata/connect:4.92.0.
  # Per-Pipeline `.spec.image` still wins if it's set; see
  # `TestRender_Deployment_ImagePrecedence` for the three-tier contract.
  cluster:
    clusterRef: { name: redpanda }
  userRef:
    name: mysql-cdc-orders-svc
  valueSources:
    - name: MYSQL_PASSWORD
      source:
        secretKeyRef: { name: mysql-cdc-creds, key: password }
    - name: MYSQL_HOST
      source:
        secretKeyRef: { name: mysql-cdc-creds, key: host }
    - name: MYSQL_USER
      source:
        secretKeyRef: { name: mysql-cdc-creds, key: username }
  configYaml: |
    input:
      mysql_cdc:
        dsn: "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:3306)/shop"
        tables: [orders]
        stream_snapshot: true
        flavor: mysql
        checkpoint_cache: mysql_cdc_orders_checkpoint
        checkpoint_key: mysql_cdc_orders_checkpoint
        checkpoint_limit: 1024
    pipeline:
      processors:
        - mapping: |
            root = this
            root.cdc_received_at = now()
    output:
      redpanda_common:                     # ← shared-client output; uses the operator-generated `redpanda:` block
        topic: "mysql.shop.orders"
        key: '${! @table }'
    cache_resources:
      - label: mysql_cdc_orders_checkpoint
        memory: { default_ttl: 1h }

Rendered connect.yaml the operator writes to the pod's ConfigMap:

cache_resources:
- label: mysql_cdc_orders_checkpoint
  memory:
    default_ttl: 1h
input:
  mysql_cdc:
    checkpoint_cache: mysql_cdc_orders_checkpoint
    checkpoint_key: mysql_cdc_orders_checkpoint
    checkpoint_limit: 1024
    dsn: ${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:3306)/shop
    flavor: mysql
    snapshot_max_batch_size: 100
    stream_snapshot: true
    tables:
    - orders
output:
  redpanda_common:
    key: ${! @table }
    topic: mysql.shop.orders
pipeline:
  processors:
  - mapping: |
      root = this
      root.cdc_received_at = now()
redpanda:                                  # ← generated from cluster.clusterRef + userRef
  sasl:
  - mechanism: SCRAM-SHA-512
    password: ${REDPANDA_SASL_PASSWORD}    # injected env from User CR's password Secret
    username: ${REDPANDA_SASL_USERNAME}    # injected env (literal: "mysql-cdc-orders-svc")
  seed_brokers:
  - redpanda-0.redpanda.redpanda.svc.cluster.local.:9093

Pod env vars projected by the operator:

• REDPANDA_SASL_USERNAME = mysql-cdc-orders-svc (literal, from User.metadata.name)
• REDPANDA_SASL_MECHANISM = SCRAM-SHA-512
• REDPANDA_SASL_PASSWORD → secretKeyRef: the User CR's password Secret
• MYSQL_PASSWORD / MYSQL_HOST / MYSQL_USER → secretKeyRef: mysql-cdc-creds (the ESO-materialized Secret backed by AWS Secrets Manager)

The Pipeline CR contains zero plaintext credentials.

Reproduction steps

1. Prerequisites

• aws + kubectl + helm + Docker on the laptop.
• An EKS-capable AWS account; AWS_PROFILE set.
• A Redpanda enterprise license file with the CONNECT product entitlement (the Connect runtime gates mysql_cdc on it).
• The PR 1337 branch checked out locally so helm install can use the operator chart from the branch.

2. Terraform (AWS infra)

The pr1337-eks-rds/terraform/ directory provisions:

• VPC (3 AZs, NAT, private subnets for EKS + RDS)
• EKS 1.31 with OIDC provider
• 1× m5.large managed node group
• RDS MySQL 8.0 db.t3.medium in private subnets, with a custom mysql8.0 parameter group enabling binlog_format=ROW + binlog_row_image=FULL (NB: GTID was dropped — RDS rejects setting gtid_mode=ON directly, requires the state-walk OFF→OFF_PERMISSIVE→ON_PERMISSIVE→ON. mysql_cdc works fine on file-position binlog reading without GTID.)
• AWS Secrets Manager secrets for the RDS master + CDC user
• IAM role for ESO with secretsmanager:GetSecretValue on those secrets, trust on the EKS OIDC provider scoped to system:serviceaccount:external-secrets:external-secrets
• ECR repo for the operator image

cd pr1337-eks-rds/terraform
terraform init
terraform plan -out tfplan
terraform apply tfplan

3. Build + push the operator image

cd <redpanda-operator repo at e7a2da66>
BUILD_GOOS=linux BUILD_GOARCH=amd64 go build -C ./operator \
  -o ../.build/redpanda-operator-linux-amd64 \
  -ldflags '...' ./cmd/main.go
BUILD_GOOS=linux GOARCH=amd64 go build -C ./alias \
  -o ../.build/alias-linux-amd64 \
  -ldflags '-X "main.AliasTo=/redpanda-operator run"' .

aws ecr get-login-password --region us-east-2 | \
  docker login --username AWS --password-stdin \
  605419575229.dkr.ecr.us-east-2.amazonaws.com

docker buildx build --platform linux/amd64 --provenance false --sbom false \
  --file operator/Dockerfile --target=manager \
  --tag 605419575229.dkr.ecr.us-east-2.amazonaws.com/redpanda-operator-pr1337:e7a2da66 \
  --push .build

4. cert-manager + ESO + EBS CSI

aws eks update-kubeconfig --region us-east-2 --name rp-pr1337-eks-rds
kubectl create namespace redpanda

# cert-manager
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.17.2/cert-manager.yaml
kubectl -n cert-manager wait --for=condition=Available deploy --all --timeout=5m

# ESO with IRSA
helm repo add external-secrets https://charts.external-secrets.io && helm repo update
helm upgrade --install external-secrets external-secrets/external-secrets \
  --namespace external-secrets --create-namespace \
  --set installCRDs=true \
  --set 'serviceAccount.annotations.eks\.amazonaws\.com/role-arn=arn:aws:iam::605419575229:role/rp-pr1337-eks-rds-eso' \
  --wait

# EBS CSI driver — EKS 1.31 doesn't auto-install it
aws iam attach-role-policy --role-name <node-role> \
  --policy-arn arn:aws:iam::aws:policy/service-role/AmazonEBSCSIDriverPolicy
aws eks create-addon --cluster-name rp-pr1337-eks-rds --region us-east-2 \
  --addon-name aws-ebs-csi-driver --resolve-conflicts OVERWRITE
kubectl patch storageclass gp2 -p '{"metadata":{"annotations":{"storageclass.kubernetes.io/is-default-class":"true"}}}'

5. Operator (PR 1337 chart)

kubectl -n redpanda create secret generic redpanda-license \
  --from-file=license=/path/to/redpanda.license

helm upgrade --install redpanda-operator <pr-1337 repo>/operator/chart \
  --namespace redpanda \
  --set image.repository=605419575229.dkr.ecr.us-east-2.amazonaws.com/redpanda-operator-pr1337 \
  --set image.tag=e7a2da66 \
  --set image.pullPolicy=IfNotPresent \
  --set rbac.createAdditionalControllerCRs=false \
  --set connectController.enabled=true \
  # NEW chart value (commit 002fd26f): chart-level default Connect image.
  # Every Pipeline CR that doesn't pin its own .spec.image inherits this,
  # so the Connect runtime version is controlled at install time rather
  # than per-CR. Per-Pipeline .spec.image still wins; if neither is set,
  # the operator falls back to the PipelineDefaultImage constant baked
  # into the binary.
  --set connectController.image.repository=docker.redpanda.com/redpandadata/connect \
  --set connectController.image.tag=4.92.0 \
  --set enterprise.licenseSecretRef.name=redpanda-license \
  --set enterprise.licenseSecretRef.key=license \
  --set crds.enabled=true --set crds.experimental=true \
  --wait

6. Redpanda CR (SASL on, 1 broker)

kubectl -n redpanda create secret generic redpanda-bootstrap-user \
  --from-literal=password=$(openssl rand -hex 16)
kubectl -n redpanda create secret generic users --from-literal=placeholder=ignored

# 01-redpanda.yaml: clusterSpec.auth.sasl.enabled=true, bootstrapUser→redpanda-bootstrap-user,
# users[]: empty (users come from User CR), tls disabled for simplicity.
kubectl apply -f manifests/01-redpanda.yaml
kubectl -n redpanda wait redpanda/redpanda --for=condition=Ready --timeout=10m

7. ESO sync of the RDS password

# 02-eso.yaml: ClusterSecretStore + ExternalSecret pulling rp-pr1337-eks-rds/cdc-user from
# AWS Secrets Manager into a K8s Secret "mysql-cdc-creds" with keys {password, host, username}.
sed -e 's|REPLACE_REGION|us-east-2|g' \
    -e 's|REPLACE_CDC_SECRET_NAME|rp-pr1337-eks-rds/cdc-user|g' \
    manifests/02-eso.yaml | kubectl apply -f -
kubectl -n redpanda wait --for=condition=Ready externalsecret/mysql-cdc-creds --timeout=2m

8. Bootstrap RDS MySQL

kubectl apply -f manifests/05-mysql-bootstrap-job.yaml
kubectl -n redpanda wait --for=condition=Complete job/mysql-bootstrap --timeout=5m
# RDS doesn't grant RELOAD; grant the MySQL 8.0 dynamic privilege FLUSH_TABLES + LOCK TABLES instead
kubectl apply -f manifests/06-grant-flush-tables.yaml
kubectl apply -f manifests/07-grant-lock-tables.yaml

9. User CR + Pipeline CR

# 03-pipeline-user.yaml: User CR scoped to write topic mysql.shop.orders + Group prefix "mysql-cdc-orders"
# 04-pipeline.yaml: Pipeline CR with cluster.clusterRef + userRef + valueSources (see above)
kubectl apply -f manifests/03-pipeline-user.yaml
kubectl -n redpanda wait --for=condition=Synced user/mysql-cdc-orders-svc --timeout=2m
kubectl apply -f manifests/04-pipeline.yaml
kubectl -n redpanda wait --for=condition=Ready pipeline/mysql-cdc-orders --timeout=5m

10. Verify CDC

PASS=$(kubectl -n redpanda get secret redpanda-bootstrap-user -o jsonpath='{.data.password}' | base64 -d)

# 5 snapshot rows
kubectl -n redpanda exec redpanda-0 -c redpanda -- \
  rpk topic consume mysql.shop.orders -n 5 --offset start \
    --user kubernetes-controller --password "$PASS" \
    --sasl-mechanism SCRAM-SHA-256 \
    --brokers redpanda-0.redpanda.redpanda.svc.cluster.local.:9093

# Live insert
kubectl apply -f - <<EOF2
apiVersion: batch/v1
kind: Job
metadata: { name: mysql-insert-frank, namespace: redpanda }
spec:
  template:
    spec:
      restartPolicy: OnFailure
      containers:
        - name: ins
          image: mysql:8.0
          env:
            - { name: MYSQL_HOST, valueFrom: { secretKeyRef: { name: rds-master-bootstrap, key: host } } }
            - { name: MYSQL_USER, valueFrom: { secretKeyRef: { name: rds-master-bootstrap, key: username } } }
            - { name: MYSQL_PWD,  valueFrom: { secretKeyRef: { name: rds-master-bootstrap, key: password } } }
          command: ["mysql","-h","\$(MYSQL_HOST)","-u","\$(MYSQL_USER)","-e","USE shop; INSERT INTO orders (customer,product,qty) VALUES ('frank','gear',11);"]
EOF2

# Verify frank appears
kubectl -n redpanda exec redpanda-0 -c redpanda -- \
  rpk topic consume mysql.shop.orders -n 1 --offset 10 \
    --user kubernetes-controller --password "$PASS" \
    --sasl-mechanism SCRAM-SHA-256 \
    --brokers redpanda-0.redpanda.redpanda.svc.cluster.local.:9093

Notes / friction points worth knowing about

1. output.redpanda vs output.redpanda_common. Connect 4.92.0's output.redpanda plugin does NOT use the top-level redpanda: block as its shared client — it requires inline seed_brokers. The plugin that DOES use the shared client is output.redpanda_common (and input.redpanda_common). The operator's auto-generated top-level redpanda block matches what redpanda_common expects. Pipelines authored against clusterRef + userRef need to use the _common variant on their input/output. Worth a docstring callout in Pipeline.spec.configYaml's godoc.

2. RDS / mysql_cdc privilege quirks.

   • RELOAD is restricted on RDS (master-only). mysql_cdc's default path runs FLUSH TABLES X WITH READ LOCK, which used to require RELOAD. Grant the MySQL 8.0 dynamic privilege FLUSH_TABLES instead — RDS allows it and mysql_cdc accepts it.
   • LOCK TABLES on the database is also needed (not just FLUSH_TABLES) — GRANT LOCK TABLES ON shop.* TO cdc@'%'.
   • GTID mode in the param group is awkward on RDS (multi-step state machine). Skip it; position-based binlog reading works.

3. EKS 1.31 doesn't bundle the EBS CSI driver. PVC creation hangs with pod has unbound immediate PersistentVolumeClaims until the addon + IAM policy are added. Step 4 above handles it.

4. Operator stale broker IP. When the Redpanda cluster rolls (e.g. flipping SASL on after first install), the operator's cached broker connection points at the old pod IP. Without a restart, the User controller's reconcile fails with dial tcp X: connect: no route to host and User CRs sit at Synced=False. Workaround: kubectl rollout restart deploy/redpanda-operator -n redpanda after the Redpanda pod is back up. Probably worth a client-connection refresh on broker-pod-change events in the operator.

5. Auto-generated redpanda: block requires the underlying cluster to actually have SASL on. The operator wires userRef → SASL config into the rendered redpanda block regardless of whether the Redpanda cluster has auth.sasl.enabled=true. If the cluster doesn't, Connect's SASL handshake gets ILLEGAL_SASL_STATE and the shared client never initializes. The operator could reject the CR (or fall through to a no-SASL render) when the resolved cluster's auth isn't on; today it silently produces a broken config. Probably worth a status condition like ClusterAuthCompatible.

6. IRSA target. userRef + AWS Secrets Manager interplay: the pipeline pod itself doesn't talk to AWS. ESO does, on the pipeline's behalf, using its own IRSA-annotated service account. The pipeline pod just reads the K8s Secret that ESO materialized. That's the standard production K8s-RDS pattern — IRSA gates AWS API access, not MySQL handshake. For native RDS IAM database auth (where the pod calls rds:GenerateDBAuthToken and uses the token as the MySQL password) the Pipeline CR would need a token-refresh primitive — that's a separate follow-up not blocking this PR.

Update 2026-05-16 — Re-test on EKS 1.34 with native RDS IAM database authentication

Re-ran the e2e against EKS 1.34 using native RDS IAM database authentication via mysql_cdc.aws.enabled: true (per the Connect docs) instead of ESO-fetched MySQL passwords. This also surfaced a deprecation issue and a new design recommendation for the Pipeline CRD.

Result: ✅ Pass with native IAM auth

• No MySQL password lives anywhere in the pipeline pod or its env. The cdc user on RDS is created with IDENTIFIED WITH AWSAuthenticationPlugin AS 'RDS'. The pipeline pod's IRSA-assumed role (...:pipeline-rds) is granted rds-db:connect on arn:aws:rds-db:<region>:<acct>:dbuser:<DbiResourceId>/cdc. mysql_cdc.aws.enabled: true makes Connect call rds:GenerateDBAuthToken and use the 15-min token as the MySQL password.
• CDC topic mysql.shop.orders populated with snapshot rows; verified via one-shot rpk consumer:

  offset=0 key=orders val={"cdc_received_at":"…","created_at":"…","customer":"alice","id":1,"product":"wrench","qty":3}
  offset=1 …carol/sprocket
  offset=2 …eve/cog
  …
  

• Used the non-deprecated redpanda output (not redpanda_common) — see design note below.

New / corrected friction points

N1 — redpanda_common is deprecated; CRD design is locked to it (NEW, biggest issue).
PR 1337's operator currently emits a top-level redpanda: block consumed only by the deprecated redpanda_common plugin. A Pipeline CR that uses the non-deprecated output.redpanda plugin still has to hand-write seed_brokers/sasl/tls even though all that information is already known to the operator via spec.cluster.clusterRef + spec.userRef. As Connect's deprecation cycle moves forward, the auto-generated config will start emitting warnings and eventually break.

Recommended redesign (inline-merge approach): the operator should scan the user-supplied connect.yaml, find any output.redpanda / input.redpanda blocks, and merge in seed_brokers / tls / sasl derived from the resolved cluster and user. The user writes only topic / key / etc.; the operator handles the wire-up. This keeps the Pipeline CRD k8s-native (the whole point of clusterRef + userRef) and follows the supported plugin path.

A working example of what the rendered config should look like (after the inline-merge):

output:
  redpanda:
    seed_brokers:
      - redpanda-0.redpanda.redpanda.svc.cluster.local.:9093
    sasl:
      - mechanism: SCRAM-SHA-512
        username: ${REDPANDA_SASL_USERNAME}
        password: ${REDPANDA_SASL_PASSWORD}
    topic: "mysql.shop.orders"
    key: '${! @table }'

The user authored:

output:
  redpanda:
    topic: "mysql.shop.orders"
    key: '${! @table }'

N6 (CORRECTED) — Native RDS IAM auth works today; no Pipeline CR primitive is required.
My original friction point #6 said the Pipeline CR would need a "token-refresh primitive" for native RDS IAM auth. That was wrong. mysql_cdc.aws.enabled: true makes Connect call rds:GenerateDBAuthToken itself and refreshes on its own; no operator changes needed. The only operator-side gap is no serviceAccountName field on the rendered Deployment — for now I annotated the redpanda namespace's default SA with the IRSA role ARN, which works but is awkward (every pipeline in the namespace shares that role).

Suggested follow-up: expose Pipeline.spec.podSpec.serviceAccountName (or similar) so the IRSA boundary can be per-pipeline.

N7 (NEW) — mysql_cdc.aws.endpoint requires host:port, not just host.
The IAM token signer reports "the provided endpoint is missing a port, or the provided port is invalid" if endpoint: <hostname> is set without :3306. This is a Connect-side input-validation gap, not a PR 1337 issue. Worth a docstring on mysql_cdc.aws.endpoint.

N8 (NEW) — IAM auth requires allowCleartextPasswords=1 in the DSN.
The Go MySQL driver refuses to send the IAM token (which it sees as a plaintext password) by default. Add &allowCleartextPasswords=1 to the DSN. TLS protects the token on the wire (which is fine — tls=skip-verify for the test, tls=true + RDS CA bundle for production). Worth a docs note on mysql_cdc.dsn and / or in the IAM-auth example.

N9 (NEW) — RDS bootstrap SQL: REQUIRE SSL; is not valid as a standalone statement.
MySQL requires REQUIRE SSL to be part of CREATE USER/ALTER USER, not a free-standing statement. A common pasted-from-docs gotcha. Not a PR issue.

Stack (this run)

Pipeline CR (this run; uses output.redpanda directly)

apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata: { name: mysql-cdc-orders, namespace: redpanda }
spec:
  cluster:
    clusterRef: { name: redpanda }
  userRef:
    name: mysql-cdc-orders-svc
  valueSources:
    # No MYSQL_PASSWORD — IAM auth supplies it on the fly.
    - name: MYSQL_HOST
      source: { secretKeyRef: { name: mysql-cdc-creds, key: host } }
    - name: MYSQL_USER
      source: { secretKeyRef: { name: mysql-cdc-creds, key: username } }
  configYaml: |
    input:
      mysql_cdc:
        dsn: "${MYSQL_USER}@tcp(${MYSQL_HOST}:3306)/shop?tls=skip-verify&allowCleartextPasswords=1"
        aws:
          enabled: true
          region: us-east-2
          endpoint: "${MYSQL_HOST}:3306"
        stream_snapshot: true
        tables: [orders]
        flavor: mysql
        checkpoint_cache: mysql_cdc_orders_checkpoint
        checkpoint_key: mysql_cdc_orders_checkpoint
        checkpoint_limit: 1024
    pipeline:
      processors:
        - mapping: |
            root = this
            root.cdc_received_at = now()
    output:
      # Using non-deprecated `redpanda`. Until the operator inline-merges
      # cluster connection fields (N1 above), seed_brokers + SASL still
      # need to be hand-written here.
      redpanda:
        seed_brokers:
          - redpanda-0.redpanda.redpanda.svc.cluster.local.:9093
        sasl:
          - mechanism: ${REDPANDA_SASL_MECHANISM}
            username: ${REDPANDA_SASL_USERNAME}
            password: ${REDPANDA_SASL_PASSWORD}
        topic: "mysql.shop.orders"
        key: '${! @table }'
    cache_resources:
      - label: mysql_cdc_orders_checkpoint
        memory: { default_ttl: 1h }
  replicas: 1